package com.starry.push.netty.websocket;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.starry.push.netty.domain.LogWsConnect;
import com.starry.push.netty.service.LogWsConnectService;
import com.starry.push.netty.util.SpringUtils;
import com.starry.push.netty.vo.Connect;
import com.starry.push.netty.vo.WsResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author WuChao
 * @version 1.0
 * @date 2020/11/3 12:12
 */
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    /**
     * channel集合,传入全局事件执行器
     */
    public static final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    public static final ConcurrentHashMap<String, Channel> onlineGroup = new ConcurrentHashMap<>();
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
        Channel channel = ctx.channel();
        String text = textWebSocketFrame.text();
        log.info("接收到的数据为{}", text);

        DataPacket dataPacket = JSONObject.parseObject(text, DataPacket.class);
        Integer action = dataPacket.getAction();
        if (ObjectUtils.equals(action, MsgActionEnum.CONNECT.type)) {
            //注册链接
            registerConnect(channel, dataPacket);
        } else if (ObjectUtils.equals(action, MsgActionEnum.KEEPALIVE.type)) {
            //心跳
            keepAlive(channel);
        }else if (ObjectUtils.equals(action, MsgActionEnum.CLIENT_TEXT.type)) {
            //收到文本消息 目前不处理
        }


//        ctx.writeAndFlush(new TextWebSocketFrame("你好"));

    }
    private void keepAlive(Channel currentChannel) {
        log.info("收到来自channel为" + currentChannel.id().asLongText() + "的心跳包");
    }
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("新接入连接id:{}", ctx.channel().id());
        channelGroup.add(ctx.channel());
        onlineGroup.put(ctx.channel().id().asLongText(), ctx.channel());
        //记录链接日志 已连接未认证
        LogWsConnectService connectService = SpringUtils.getBean(LogWsConnectService.class);
        connectService.save(LogWsConnect.builder()
                .channelId(ctx.channel().id().asLongText())
                .status(0)
                .createTime(LocalDateTime.now())
                .serverHost(ctx.channel().remoteAddress().toString())
                .build());
        log.info("当前客户端连接数：{}", channelGroup.size());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        //长id唯一
        String channelId = ctx.channel().id().asLongText();
        log.info("客户端移除,channelId为：" + channelId);
        //日志移除
        removeWsConnectLog(ctx.channel());
        //这步可以不用做 断开连接的 channelGroup 会自动移除
        channelGroup.remove(ctx.channel());
        onlineGroup.remove(ctx.channel().id().asLongText(), ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("出现异常：" + cause.getMessage());

        ctx.channel().close();
        //日志移除
        removeWsConnectLog(ctx.channel());
        channelGroup.remove(ctx.channel());
        onlineGroup.put(ctx.channel().id().asLongText(), ctx.channel());
    }


    private void registerConnect(Channel channel, DataPacket dataPacket) {
        RedisTemplate redisTemplate = (RedisTemplate) SpringUtils.getBean("redisTemplate");
        LogWsConnectService wsConnectService = SpringUtils.getBean(LogWsConnectService.class);
        String token = dataPacket.getToken();
        if (StringUtils.isBlank(token)) {
            channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(WsResponse.error("token无效 注册失败"))));
        }
        Boolean key = redisTemplate.hasKey(token);
        if (!key) {
            log.info("{} ws链接校验失败", token);
            channelGroup.remove(channel);
            onlineGroup.remove(channel.id().asLongText(),channel);
            channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(WsResponse.error("token无效 注册失败"))));
            channel.close();
        }else {
            log.info("注册成功了,链接地址:{},Id:{}", channel.remoteAddress(), channel.id().asLongText());
            String jsonData = dataPacket.getJsonData();
            if (StringUtils.isBlank(jsonData)) {
                channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(WsResponse.error("附加消息为空 注册失败"))));
                return;
            }
            redisTemplate.delete(token);
            Connect connect = JSONObject.parseObject(jsonData, Connect.class);
            wsConnectService.save(LogWsConnect.builder()
                    .serverHost(channel.remoteAddress().toString())
                    .createTime(LocalDateTime.now())
                    .status(1)
                    .channelId(channel.id().asLongText())
                    .uniqueIdentification(connect.getUniqueIdentification())
                    .clientType(connect.getClientType())
                    .build()
            );
            wsConnectService.remove(new QueryWrapper<LogWsConnect>()
                    .eq(LogWsConnect.COL_CHANNEL_ID, channel.id().asLongText())
                    .eq(LogWsConnect.COL_STATUS, 0)
            );
        }
    }

    private void removeWsConnectLog(Channel channel) {
        LogWsConnectService wsConnectService = SpringUtils.getBean(LogWsConnectService.class);
        List<LogWsConnect> wsConnects = wsConnectService.list(new QueryWrapper<LogWsConnect>()
                .eq(LogWsConnect.COL_CHANNEL_ID, channel.id().asLongText()));
        if (CollectionUtils.isNotEmpty(wsConnects)) {
            wsConnects.forEach(logWsConnect -> logWsConnect.setStatus(3));
            wsConnectService.updateBatch(wsConnects);
        }
    }
}
